package com.cloud.point.concurrent.aeron;

import io.aeron.Aeron;
import io.aeron.Publication;
import io.aeron.Subscription;
import io.aeron.driver.MediaDriver;
import io.aeron.logbuffer.FragmentHandler;

import java.nio.ByteBuffer;

import lombok.extern.slf4j.Slf4j;
import org.agrona.concurrent.IdleStrategy;
import org.agrona.concurrent.SleepingIdleStrategy;
import org.agrona.concurrent.UnsafeBuffer;

/**
 * TestAeron
 *
 * @author 24373
 * @version 1.0
 * @date 2024-06-28 16:17
 */
@Slf4j
public class TestAeron {

    public static void main(String[] args) {
        String channel = "aeron:ipc";
        String message = "this is a new Message.";
        IdleStrategy idleStrategy = new SleepingIdleStrategy();
        UnsafeBuffer unsafeBuffer = new UnsafeBuffer(ByteBuffer.allocate(256));

        try (
                MediaDriver driver = MediaDriver.launch();
                Aeron aeron = Aeron.connect();
                Subscription sub = aeron.addSubscription(channel, 10);
                Publication pub = aeron.addPublication(channel, 10);
        ) {
            while (!pub.isConnected()) {
                // 连接失败，阻塞
                idleStrategy.idle();
            }
            unsafeBuffer.putStringAscii(0, message);
            log.info("sending:" + message);
            // 尝试发布
            while (pub.offer(unsafeBuffer) < 0) {
                idleStrategy.idle();
            }
            FragmentHandler handler = ((buffer, offset, length, header) -> {
                log.info("received:" + buffer.getStringAscii(offset));
            });
            while (sub.poll(handler, 1) <= 0) {
                idleStrategy.idle();
            }
        }

    }

}
